package com.mng.window;

import com.mng.dto.SensorReading;
import org.apache.flink.api.common.functions.AggregateFunction;
import org.apache.flink.api.common.functions.MapFunction;
import org.apache.flink.api.java.functions.KeySelector;
import org.apache.flink.streaming.api.datastream.DataStreamSource;
import org.apache.flink.streaming.api.datastream.SingleOutputStreamOperator;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.streaming.api.windowing.assigners.TumblingProcessingTimeWindows;
import org.apache.flink.streaming.api.windowing.time.Time;

public class windowTest {
    public static void main(String[] args) throws Exception {
        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
        DataStreamSource<String> stringDs = env.socketTextStream("localhost",7777);

        SingleOutputStreamOperator<SensorReading> sensorDs = stringDs.map(line ->{
            String[] strings = line.split(",");
            return new SensorReading(strings[0],Long.parseLong(strings[1]),Double.parseDouble(strings[2]));
            }
        );
        sensorDs.keyBy(sensor -> sensor.getId())
                .window(TumblingProcessingTimeWindows.of(Time.seconds(30)))
                .aggregate(new AggregateFunction<SensorReading, SensorReading, SensorReading>() {
                    //新建的累加器
                    @Override
                    public SensorReading createAccumulator() {
                        return null;
                    }

                    //每个数据在上次的基础上累加
                    @Override
                    public SensorReading add(SensorReading sensorReading, SensorReading o) {
                        if(o == null || sensorReading.getTemperaturre()>o.getTemperaturre()) {
                            return sensorReading;
                        }
                        return o;
                    }

                    //返回结果值
                    @Override
                    public SensorReading getResult(SensorReading o) {
                        return o;
                    }

                    // 分区合并结果(TimeWindow一般用不到，SessionWindow可能需要考虑合并)
                    @Override
                    public SensorReading merge(SensorReading o, SensorReading acc1) {
                        return null;
                    }
                })
                .print();
                //.max("temperaturre")
                //.print();
        env.execute();
    }
}
